package com.cc.zk.distribute.idgen;

import com.cc.zk.mpsc.ServiceThread;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.retry.RetryNTimes;

import java.util.concurrent.atomic.AtomicLong;

/**
 * zookeeper 生成有序的id
 *
 * @author cc
 * @date 2022/8/24
 */
@Slf4j
public class ZkSequenceIdGenerator implements IdGenerator {

    /**
     * 持久化id 的 path
     */
    public static final String SEQUENCE_PATH = "/push-center/sequence/";
    public static final String WORKER_PATH = "/push-center/zkId/worker-";
    /**
     * 服务实例的IdWorker
     */
    private final IdWorker worker;
    /**
    * 服务名
    */
    private final String type;

    /**
     * 该服务实例的worker 节点 id
     */
    private long workerId;
    private long workerIdPart;
     /**
     * 可以生成最大id
     */
    private long maxSeq;

    private volatile boolean inited = false;

    /**
     * 2的7次方-1  同一个微服务127个实例，可以根据实际情况适当调整这个值
     */
    private static final long WORKER_ID_BITS = 8L;
    /**
     * 100w个10亿
     */
    private static final long SEQUENCE_BITS = 56L;

    /**
     * 每次生成id使用这个值做自增长
     * 问题：服务重启后  序号不能从0 开始，如何解决？持久化。id异步持久化机制
     */
    private volatile AtomicLong localSequence = new AtomicLong(0);
    /**
     * 每次生成id 会累加1，做持久化时使用
     */
    private volatile AtomicLong delta = new AtomicLong(0);

    /**
     * 异步持久化服务类，mpsc
     */
    private final RemoteIncreaseService remoteIncreaseService = new RemoteIncreaseService();
    /**
     * 最大的 worker id ，256
     * -1 的补码（二进制全1）右移13位, 然后取反
     */
    private final static long MAX_WORKER_ID = ~(-1L << WORKER_ID_BITS);

    RetryPolicy retryPolicy = new RetryNTimes(10, 100);

    /**
     * zk分布式 自增id
     */
    DistributedAtomicLong remoteSequence;

    public ZkSequenceIdGenerator(String type) {
        this.type = type;
        worker = IdWorker.getInstance();
    }

    /**
     * 初始化单例
     * <p>
     * workerId 节点Id,最大256
     */
    private synchronized void init() {
        workerId = worker.getId();
        if (workerId > MAX_WORKER_ID) {
            // zk分配的workerId过大
            throw new IllegalArgumentException("worker Id wrong: " + workerId);
        }
        String sequenceZnodeName = SEQUENCE_PATH + type + "/worker-" + workerId;
        remoteSequence = new DistributedAtomicLong(worker.getClient(), sequenceZnodeName, retryPolicy);
        // 做一下校验，把数据库里边的最大值取出来，max，和remoteSequence值做比对
        localSequence.set(getRemoteValue());
        //workerId 左移
        this.workerIdPart = (this.workerId + 1) << SEQUENCE_BITS;
        maxSeq = ~(-1L << SEQUENCE_BITS);
        log.info("worker id is {},worker id part is {}", workerId, workerIdPart);
        remoteIncreaseService.start();
        inited = true;
    }

    @Override
    public Long nextId() {
        if (!inited) {
            init();
        }
        return generateId();
    }


    private Long generateId() {
        long sequence = localSequence.incrementAndGet();
        if (sequence > maxSeq) {
            throw new RuntimeException("id 生产出错，太大了,>" + maxSeq);
        }
        delta.incrementAndGet();
        remoteIncreaseService.wakeup();
        //将本地的工作id和生成的有序id进行或运算 生成最终的id
        return workerIdPart | sequence;
    }


    /**
     * 获取远程值
     *
     * @return {@link Long}
     */
    public Long getRemoteValue() {
        AtomicValue<Long> sequence;
        try {
            sequence = this.remoteSequence.get();

            if (sequence.succeeded()) {
                Long val = sequence.postValue();
                log.debug("生产id 上一次的值：{}", val);
                return val;
            } else {
                throw new RuntimeException("id 生产出错");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        throw new RuntimeException("id 生产出错");

    }


    /**
     * 进行id持久化
     * 问题：在持久化的时候发生了异常。可能没有持久化成功。如何解决？
     * 解决思路：在id启动时候 做个校验，吧数据 取出来 和remoteSequence值做比对。
     */
    public void incrementRemote() {
        while (true) {
            //先算出 增量的id
            long toAdd = delta.get();
            if (toAdd <= 0) {
                break;
            }
            AtomicValue<Long> sequence = null;
            try {
                //持久化是低性能的。比如本地增长了100个id，这里持久化一次就可以了。
                sequence = this.remoteSequence.add(toAdd);
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (sequence.succeeded()) {
                log.error("生产id 的seq：{}", sequence.postValue());
                //减去 增加的值，如果还有就继续同步
                delta.addAndGet(-toAdd);
                break;
            } else {
                log.error("生产id 失败,继续重试：{}", type);
            }
        }
    }


    /**
     * 远程id持久化
     * mpsc
     *
     * @author cc
     * @date 2022/08/24
     */
    class RemoteIncreaseService extends ServiceThread {

        @Override
        public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped()) {
                this.waitForRunning(1000);
                ZkSequenceIdGenerator.this.incrementRemote();
            }
            log.info(this.getServiceName() + " service end");
        }

        @Override
        public String getServiceName() {
            return "RemoteIncreaseService:" + type;
        }
    }
}
